package com.bw.medical.realtime.dwd.app;

import com.bw.medical.realtime.common.base.BaseSQL;
import com.bw.medical.realtime.common.constant.Constant;
import com.bw.medical.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdUserPatientAdd extends BaseSQL {

    public static void main(String[] args) {
        new DwdUserPatientAdd().start(
                10016,
                4,
                "medical_dwd_user_patient_add"
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tabEnv) {

        readOdsDb(tabEnv,"medical_dwd_user_patient_add");

        Table addUTable = tabEnv.sqlQuery("select\n" +
                "`data`['id'] id, \n" +
                "`data`['create_time'] add_time, \n" +
                "`data`['birthday'] birthday, \n" +
                "`data`['gender'] gender, \n" +
                "concat(substr(`data`['name'],1,1),'*') `name`, \n" + //对用户信息进行脱敏
                "`data`['user_id'] user_id\n" +
                "from \n" +
                "topic_db\n" +
                "where `table` = 'patient'\n" +
                "and `type` = 'insert'");

        tabEnv.createTemporaryView("add_us_table",addUTable);

        tabEnv.executeSql("create table " + Constant.DWD_USER_PATIENT_ADD+ "(\n" +
                "`id` string,\n" +
                "`add_time` string, \n" +
                "`birthday` string,\n" +
                "`gender_code` string,\n" +
                "`name` string,\n" +
                "`user_id` string\n" +
                ") " + SQLUtil.getKafkaSinkSQL(Constant.DWD_USER_PATIENT_ADD));

        addUTable.executeInsert(Constant.DWD_USER_PATIENT_ADD);
//        addUTable.execute().print();
    }
}
